package com.max.spark.redis

import com.max.spark.utils.ConfUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

// 隐式转换可以调出的读取Redis的方法，都是以fromRedis开头的，都是redis可以存储的数据结构
import com.redislabs.provider.redis._

object SparkRedisApp {
  val confPath = "conf/redis.properties"
  val hostname = ConfUtil.getProperty(confPath, "redis.hostname")
  val port = ConfUtil.getProperty(confPath, "redis.port")
  val username = ConfUtil.getProperty(confPath, "redis.username")
  val password = ConfUtil.getProperty(confPath, "redis.password")

  def main(args: Array[String]): Unit = {
    val conf: SparkConf =new SparkConf().setAppName("setRedis").setMaster("local[2]")
    conf.set("redis.host", hostname)    //redis 主机节点
    conf.set("redis.port", port)  //端口号，不填默认为6379
    conf.set("redis.auth","null")  //用户权限配置
    conf.set("redis.db","0")  //数据库设置
    conf.set("redis.timeout","2000")  //设置连接超时时间

    val session: SparkSession =SparkSession.builder().config(conf).getOrCreate()
    val sc: SparkContext =session.sparkContext

    // 读取
    val rdd = sc.fromRedisKV("a")
    rdd.collect().foreach(println)

    // 写入
    val data: Seq[(String, String)] = Seq[(String,String)](("high","111"), ("abc","222"), ("together","333"))
    val redisData:RDD[(String,String)] = sc.parallelize(data)
    sc.toRedisKV(redisData)
  }
}
